package cn.ac.iie.ulss.indexer.worker;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;

public class RocketdataHandler implements MessageListenerConcurrently {

    public static Logger log = Logger.getLogger(RocketdataHandler.class.getName());
    private ArrayBlockingQueue<MessageExt> inbuffer;

    public RocketdataHandler(ArrayBlockingQueue<MessageExt> buf) {
        this.inbuffer = buf;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            for (MessageExt msg : msgs) {                
                while (!this.inbuffer.offer(msg, 300L, TimeUnit.MILLISECONDS)) {
                    log.warn("put message to buffer fail,will retry for " + msg.getTopic());
                }
            }
        } catch (Exception e) {
            log.error(e, e);
        } finally {
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}